package com.zdxlz.fcmp.common.mqtt.handler;

import com.zdxlz.fcmp.common.core.util.JsonUtil;
import com.zdxlz.fcmp.common.mqtt.dto.IctMqttResponse;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.springframework.core.task.TaskExecutor;

import java.util.Map;
import java.util.concurrent.CompletableFuture;

/**
 * @author: aman
 * @date 2025/2/11 9:34
 * @description
 */

@Slf4j
public class CustomMqttCallbackHandler extends BaseMqttCallbackHandler {

    private final TaskExecutor taskExecutor;
    private final Map<String, CompletableFuture<IctMqttResponse<?>>> pendingResponses;

    public CustomMqttCallbackHandler(TaskExecutor taskExecutor, Map<String, CompletableFuture<IctMqttResponse<?>>> pendingResponses) {
        this.taskExecutor = taskExecutor;
        this.pendingResponses = pendingResponses;
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        taskExecutor.execute(() -> {
            String payload = new String(message.getPayload());
            log.debug("收到消息，topic：{}，数据：{}", topic, payload);

            // 处理设备响应
            CompletableFuture<IctMqttResponse<?>> future = pendingResponses.remove(topic);
            if (future != null) {
                // 解析payload并设置到response中
                IctMqttResponse response = JsonUtil.fromJson(payload, IctMqttResponse.class);
                response.setTopic(topic);
                future.complete(response);
            }
        });
    }

}
